package dyyx;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicInteger;

import redis.clients.jedis.util.Hashing;

public class SimpleConsistHashDemo {
	
	public static void main(String[] args) throws Exception {
		
		System.out.println("hello,SimpleConsistHashDemo");
		
		List<Node> nodes = new ArrayList<Node>();
		int nodeNum = 3;
		for(int i=0;i<nodeNum;i++){
			nodes.add(new Node("node"+i));
		}
		
		int virtualNodeNum = 160;
		
		
		TreeMap<Long,Node> map = new TreeMap<Long,Node>();
		
		Node node = null;
		String key = null;
		
		// hash 函数 ，使用 jedis 里的
		//final Hashing hashfunction = Hashing.MD5;
		final Hashing hashfunction = Hashing.MURMUR_HASH;
		
		// 构建 节点 hash 环 
		for(int i=0;i<nodeNum;i++){
			node = nodes.get(i);
			
			for(int j=0;j<virtualNodeNum;j++){
				key = node.name +"_"+j;
				map.put(hashfunction.hash(key), node);
			}
		}
		
		System.out.println(map);
		
		// 根据 key 查看 节点信息
		int dataKeyNum = 21;
		String dataKey = null;
		for(int i=0;i<dataKeyNum;i++){
			dataKey = "dataKey"+i;
			node = getNode(map, hashfunction, dataKey);
			System.out.println("dataKey="+dataKey+",node="+node);
		}
		
		// key 分布均衡性测试 
		Map<String,AtomicInteger> countMap = new HashMap<String,AtomicInteger>();
		dataKeyNum = 10000;
		AtomicInteger count = null;
		for(int i=0;i<dataKeyNum;i++){
			dataKey = "dataKey"+i;
			node = getNode(map, hashfunction, dataKey);
			count = countMap.get(node.name);
			if(count==null){
				count = new AtomicInteger(0);
				countMap.put(node.name, count);
			}
			count.incrementAndGet();
		}
		
		// MURMUR_HASH {node2=3144, node0=3423, node1=3433}
		// MD5         {node2=3116, node0=3244, node1=3640}
		System.out.println(countMap);
		
	
		
	}
	
	private static Node getNode(TreeMap<Long,Node> map,Hashing hashfunction,String key){
		SortedMap<Long,Node> tailMap = map.tailMap(hashfunction.hash(key));
		Long nodeKey = null;
		if(tailMap==null || tailMap.isEmpty()){
			nodeKey = map.firstKey();
		}else{
			nodeKey = tailMap.firstKey();
		}
		return map.get(nodeKey);
	}
	
	private static  class Node{
		public final String name;
		
		public Node(String name){
			this.name = name;
		}
		
		public String toString(){
			return name;
		}
	}
}


